/*
 * 文件名               :  DataTransferThread.java
 * 用途                    :  用于向流写入对象,由于使用ObjectOutputStream,有内存泄露的风险,缺省情况下每写入100个对象就reset流
 * 最后修改日期:  $LastChangedDate: 2009-09-07 13:35:47 +0000 (Mon, 07 Sep 2009) $
 * 最后修改者     :  $LastChangedBy: wendal1985@gmail.com $
 * 创建者               :  Wendal Chen <wendal1985@gmail.com>
 */
package com.wendal.java.preproxy.thread;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import com.wendal.java.preproxy.bean.MessageBean;
import com.wendal.java.preproxy.util.Toolkit;

public class DataTransferThread extends Thread {

	private Socket socket;
	
//	private Log log = LogFactory.getLog(getClass());
	
	private Queue<MessageBean> queue;
	
	public DataTransferThread(Socket socket) throws IOException {
		this.socket = socket;
		queue = new LinkedBlockingQueue<MessageBean>();
	}
	
	public void addMessageBean(MessageBean messageBean){
		synchronized (queue) {
			queue.add(messageBean);
		}
	}
	
	@Override
	public void run() {
		ObjectOutputStream oos = null;
		try {
			oos = new ObjectOutputStream(socket.getOutputStream());
		} catch (IOException e1) {
			return;
		}
		int count = 0;
		while(socket.isConnected() && ! socket.isOutputShutdown()){
			MessageBean messageBean;
			synchronized (queue) {
				messageBean = queue.poll();
			}
			if(messageBean == null){
//				System.out.println("传输队列没有找到对象,休眠100ms");
				try {
					Thread.sleep(100);
				} catch (InterruptedException e) {
					System.out.println("A thread get some exception: " + e.getMessage());
					break;
				}
			}else{
				try {
					oos.writeObject(messageBean);
					oos.flush();
					if(count > 100){
						oos.reset();
						count=0;
					}else{
						count++;
					}
//					log.info("成功传输一个对象: "+messageBean +" " + messageBean.getFrom_id());
				} catch (IOException e) {
					e.printStackTrace();
					break;
				}
			}
		}
		Toolkit.close(socket);
	}
}
